package my.thesis.bolts.persistorsReliable;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import my.thesis.analytics.metrics.LastTimePersistMetric;

import org.json.JSONObject;

import redis.clients.jedis.Jedis;
import backtype.storm.Config;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class InfluenceStreamPersistorReliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private Jedis jedis;
	private OutputCollector collector;
	
	transient LastTimePersistMetric _lastTimeMetric;
	private static Boolean metricsOn=false;

	private String database;
	
	public InfluenceStreamPersistorReliable (String connectionString)
	{
		this.database=connectionString;
	}
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
	{
		jedis=new Jedis(this.database);	
		jedis.select(6);
		this.collector=collector;
				
		if(metricsOn)
		{
			_lastTimeMetric=new LastTimePersistMetric();
			context.registerMetric("mentionPersist", _lastTimeMetric, 50);
			_lastTimeMetric.setStartTime();
		}
	}
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		// TODO Auto-generated method stub
	}
	
	public void execute(Tuple tuple) 
	{	
		
			if(metricsOn)
			{
				_lastTimeMetric.setTimeReceived();
			}
			
			//TEST
		    System.out.println("Persisting mention...");

		  	@SuppressWarnings("unchecked")
			ConcurrentHashMap<String, HashMap<String, Integer>> userMentionsFollowers  = (ConcurrentHashMap<String, HashMap<String, Integer>>) tuple.getValueByField("userMentionsFollowers");
				 
			for(Entry<String, HashMap<String, Integer>> entry : userMentionsFollowers.entrySet())	
			{
				 String mentionsUsername = entry.getKey();

				 HashMap<String, Integer> mentionsFrom = userMentionsFollowers.remove(mentionsUsername);
				 
				 for(String h : mentionsFrom.keySet())
				 {
					 JSONObject mentionedFrom = new JSONObject();
					
					 mentionedFrom.put(h, mentionsFrom.get(h));
					 
					 try
			    	 {
						 jedis.sadd(mentionsUsername, mentionedFrom.toString());
			    	 }
					 catch(Exception e)
			    	 {
			    		 System.out.println("Error saving to redis...");
						 //collector.fail(tuple);
						 return;
			    	 }
				 }	 
			}
			
			collector.ack(tuple);
	}
			
	public void cleanup()
	{
	    if (jedis.isConnected()) 
	    {
		    jedis.quit();
		}	
	}

}
